package com.ibm.cps.processors.window;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.joda.time.DateTime;

import com.fasterxml.jackson.databind.node.ArrayNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.ibm.cps.dft.FixedSizeQueue;
import com.ibm.cps.message.MessageFields;
import com.ibm.cps.message.ParsedDataMessage;
import com.ibm.cps.message.ParsedDataMessageBuilder;
import com.ibm.cps.newmessage.AbstractMetadata;
import com.ibm.cps.newmessage.MetadataInputSchema;
import com.ibm.cps.newmessage.MetadataInputSchema.FieldType;
import com.ibm.cps.processors.BasicProcessor;
import com.ibm.cps.processors.Info;
import com.ibm.cps.sql.expression.function.meta.ValueParam;
import com.ibm.datamodel.timeseries.TimeSeriesSchema;
import com.ibm.interfaces.ITimeSeriesSchema;
import com.ibm.util.DataTypes;
import com.ibm.util.ErrorCode;
import com.ibm.util.JSONUtility;
import com.ibm.util.RandomUtil;
import com.ibm.util.exception.CPSException;

/**
 * Created by telekinesis on 6/8/15.
 *     {
        "processorid": "5627347916d3b38d847a9b18",
        "groupid": "{{groupid}}",
        "parentprocessors": [
            "55fbd240be50bfdc16a9ae69"
        ],
        "type": "window",
        "name": "dtw-test_windowing",
        "step": 1,
        "size": 5
    },
    checked.
 */
@Info(Name=MessageFields.SLIDING_WINDOW_TYPE,Type=Info.ProcessorType.Other)
public class WindowProcessor extends BasicProcessor {

	@ValueParam
	private int size;
	@ValueParam
	private int step;
	
	private ITimeSeriesSchema parentSchema;
	private final Map<Comparable, FixedSizeQueue<ParsedDataMessage>> buffer = new HashMap<>();
	private ITimeSeriesSchema outputSchema;

	public WindowProcessor(String tenantId, String processorId, String source, ITimeSeriesSchema parentSchema) throws CPSException {
		super(tenantId, processorId, source, parentSchema);
		this.parentSchema = parentSchema;
	}
	
	@Override
	public void open() throws CPSException {
		this.outputSchema = createOutputSchema();
	}

	private ITimeSeriesSchema createOutputSchema() throws CPSException {
		// create an array for each field with size dimension
		ArrayNode arrayNode = JSONUtility.newArrayNode();
		byte[] types = parentSchema.getDataTypes();
		String[] names = parentSchema.getTagNames();
		for (int i = 0; i < types.length; i++) {
			if(i == parentSchema.getKeyIndex() || i == parentSchema.getTimestampIndex()){
				// skip the tskey and timeStamp
				continue;
			}
//			String fieldName = names[i]+"_array";
			String fieldName = names[i];
			String fieldType = DataTypes.getDataType(types[i]);
			
// modify by gxj to support array
			ObjectNode node = JSONUtility.newObjectNode();
			node.put(MessageFields.NAME, fieldName);
			if (fieldType.equals("int")) {
				node.put(MessageFields.DATATYPE, "int[]");
			}else if (fieldType.equals("double")) {
				node.put(MessageFields.DATATYPE, "double[]");
			}else if (fieldType.equals("string")) {
				node.put(MessageFields.DATATYPE, "string[]");
			}else {
				throw new CPSException(ErrorCode.NO_IMPLEMENTATION_CODE, "data type not support for array :" + fieldType);
			}
			arrayNode.add(node);
		}
		String schemaId = RandomUtil.getRandomId();
		return new TimeSeriesSchema(tenantId, schemaId, arrayNode, (new DateTime()).getMillis());
	}

	@Override
	public Collection<ParsedDataMessage> execute(ParsedDataMessage message) throws CPSException {
		Comparable timeseriesKey = message.getTsKey();
		ParsedDataMessage parsed = (ParsedDataMessage) message;
		FixedSizeQueue<ParsedDataMessage> bufferForTimeseries = buffer.get(timeseriesKey);
		if (bufferForTimeseries == null) {
			bufferForTimeseries = new FixedSizeQueue<>(size);
			buffer.put(timeseriesKey, bufferForTimeseries);
		}
		if (!containsNullValue(parsed)) {
			bufferForTimeseries.add(parsed);
		}
		System.out.println("==============Window input============");
		System.out.println(Arrays.toString(parsed.getObjectValues()));
		if (bufferForTimeseries.isFull()) {

			ParsedDataMessage[] output = new ParsedDataMessage[] {};
			output = bufferForTimeseries.toArray(output);
//			Comparable[] outputValues = new Comparable[outputSchema.getDataTypes().length - 2];
			Comparable tsKey = output[0].getTsKey();
			long timestamp = output[0].getTimestamp();
//			outputValues[0] = tsKey;
//			outputValues[1] = timestamp;
			
			ParsedDataMessageBuilder builder = ParsedDataMessageBuilder.createBasicBuilder(outputSchema, tsKey, timestamp);
			
			
			// each field is a vector with the number of size data elements
			// cancel to support array
//			for (int i = 2; i < parentSchema.getTagCount(); i++) {
//				String originalFieldName = parentSchema.getTagNames()[i];
//				Comparable[] _tmp = new Comparable[output.length];
//				for (int j = 0; j < output.length; j++) {
//					String outputFieldName = originalFieldName + "(" + j + ")";
//					int outputIndex = outputSchema.getTagId(outputFieldName);
//					Comparable value = output[j].getValue(i);
//					outputValues[outputIndex] = value;
//				}
//			}
			
			for (int i = 0; i < parentSchema.getTagCount(); i++) {
				String originalFieldName = parentSchema.getTagNames()[i];
				if(i == parentSchema.getKeyIndex() || i == parentSchema.getTimestampIndex()){
					// skip the tsKey and timeStamp
					continue;
				}
				Double[] _tmp = new Double[output.length];
				for (int j = 0; j < output.length; j++) {
//				int outputIndex = outputSchema.getTagId(originalFieldName);
//				int Index = parentSchema.getTagId(originalFieldName);
//				outputValues[outputIndex] = value;
				
					double value = (double) output[j].getValue(i);
					_tmp[j] = value;
					
				}
			
				builder =   builder.append(originalFieldName, _tmp);
			}

			for (int i = 0; i < step; i++) {
				bufferForTimeseries.remove();
			}
			List<ParsedDataMessage> result = new ArrayList<>();
//			ParsedDataMessage parsedData = outputSchema.buildParsedDataMessage(outputValues);
			ParsedDataMessage parsedData = builder.toParsedDataMessage();
			
			result.add(parsedData);

			System.out.println("==============Window output============");
			System.out.println(Arrays.toString(outputSchema.getTagNames()));
			System.out.println(Arrays.toString(parsedData.getObjectValues()));
			return result;
		} else {
			return null;
		}
	}

	@SuppressWarnings("rawtypes")
	private boolean containsNullValue(ParsedDataMessage message) {
		Object[] values = message.getObjectValues();
		for (Object v : values) {
			if (v == null) {
				System.out.println("WindowProcessor: received null value in " + values);
				return true;
			}
		}
		return false;
	}

	@Override
	public void dispose() {
		buffer.clear();
	}

	@Override
	public ITimeSeriesSchema getTimeSeriesSchema() throws CPSException {
		return outputSchema;
	}

}